# -*- coding: utf-8 -*-
import os
import subprocess

# import ksycopg2
import time

import boto3
import re
import paramiko as paramiko
import pymysql
import cx_Oracle
import pymssql
import psycopg2
import pymysql.cursors
import pymongo
import jaydebeapi
from boto3 import Session
from loguru import logger
from pymysql.cursors import DictCursor
from pymysql import Connection
from pyhive import hive
from elasticsearch import Elasticsearch
from ftplib import FTP
import paramiko
import dolphindb as ddb
import os
# import pyarrow.dataset as ds
from hdbcli import dbapi





class MongoDBOperate:
    def __init__(self, operate, data, table):
        self.operate = operate
        self.data = data
        self.table = table


class ElasticSearchOperate:
    def __init__(self, operate, table, data=None, hasindices=False):
        self.operate = operate
        self.data = data
        self.table = table
        self.hasindices = hasindices


class DataConnect(object):
    def __init__(self, env='online'):
        if env == 'local':
            self.mysql_ip = '127.0.0.1'
            self.ora11g_ip = '172.20.123.151'
            self.ora12c_ip = '127.0.0.1'
            self.psql_ip = '127.0.0.1'
            self.sqls_ip = '127.0.0.1'
            self.starrock_ip = '127.0.0.1'
            self.tidb_ip = '127.0.0.1'
            self.doris_ip = '8.130.66.38'
            self.iceberg_ip = '127.0.0.1'
            self.es_ip = '127.0.0.1'
            self.hive_ip = '127.0.0.1'
            self.mongodb_ip = '127.0.0.1'
            self.dolphindb_ip = '127.0.0.1'
            self.kingbase_ip = '127.0.0.1'
            self.db2_ip = '127.0.0.1'
            self.datasource01_ip = '8.130.66.38'
            self.s3redshift_ip = "s3redshift.cubsihkk6xse.cn-northwest-1.redshift.amazonaws.com.cn"
            self.ftp_ip = "8.130.94.143"
            self.sftp_ip = "8.130.94.143"
            self.informix_ip = "127.0.0.1"
            self.db2_ip = "127.0.0.1"
            self.openGauss_ip = "127.0.0.1"
            self.highGo_ip = "127.0.0.1"
            self.dm_ip = "127.0.0.1"
        elif env == 'online':
            self.selectdb_ip='selectdb-cn-m4141e5ch02-public.selectdbfe.zhangbei.rds.aliyuncs.com'
            self.mysql_ip = 'd01'
            self.ora11g_ip = '172.20.123.151'
            self.ora12c_ip = 'd01'
            self.psql_ip = '172.20.123.151'
            self.sqls_ip = 'd01'
            self.starrock_ip = 'd01'
            self.tidb_ip = 'd01'
            self.doris_ip = 'd01'
            self.iceberg_ip = 'd01'
            self.es_ip = 'd01'
            self.hive_ip = 'd01'
            self.mongodb_ip = 'd01'
            self.kingbase_ip = 'd01'
            self.datasource01_ip = 'd01'
            self.dolphindb_ip = 'd01'
            self.db2_ip = 'd01'
            self.s3redshift_ip = "s3redshift.cubsihkk6xse.cn-northwest-1.redshift.amazonaws.com.cn"
            self.informix_ip = 'd01'
            self.openGauss_ip = 'd01'
            self.highGo_ip = 'd01'
            self.db2_ip = 'd01'
            self.ftp_ip = "ws4"
            self.sftp_ip = "ws4"
            self.dm_ip = "d01"
            self.sap_hana_ip = "d01"

        self.flag_db = {
            "mysql": '', "psql": '', "sqls": '', "ora11g": '', "ora12c": '',
            "tidb": '', "doris": '', "hive": '', "iceberg": '', "kingbase": '',
            "s3redshift": '', "starrocks": '',
            "kafka": '', "es": '', "mongodb": '', "s3": '', "ftpfile": '', "sftpfile": '',
            "http": '', "localfile": '', 'ossfile': '',
            "maxcomputer": '', "db2": '', "gaussdb": '',
            "informix": '', "dameng": '', "openGauss": '', "highGo_ip": ''
        }

    def exec_iceberg_sql(self):
        pass
        # os.environ['AWS_ACCESS_KEY_ID'] = 'minio-root-user'
        # os.environ['AWS_SECRET_ACCESS_KEY'] = 'minio-root-password'
        #
        # # Minio的S3兼容接口的端点
        # os.environ['S3_ENDPOINT'] = 'http://192.168.124.199:9001'
        # os.environ['S3_USE_HTTPS'] = 'True'  # 或者 'False'，取决于你的Minio配置
        # os.environ['S3_VERIFY_SSL'] = 'False'  # 如果Minio使用自签名证书，你可能需要设置为False
        #
        # # 设置Minio上的Iceberg表路径
        # table_path = 's3://examples/initial-warehouse/0192d1d1-fd6b-77d3-9548-7ad6219fbf91/0193664f-229f-7b11-a2b6-675fc3dc0d9a'
        #
        # # 使用pyarrow.dataset读取Iceberg表
        # dataset = ds.dataset(table_path, format='iceberg')
        #
        # # 读取所有数据到pandas DataFrame
        # # 注意：如果表非常大，这可能会消耗大量内存
        # df = dataset.to_table().to_pandas()
        #
        # # 展示数据
        # print(df.head())

    def exec_selectdb_sql(self, db, sql_str):
        logger.info("执行的selectdb语句是：{}, 对应数据库是：{}".format(sql_str, db))

        response_data = {'message': '', 'data': []}
        try:
            db = pymysql.connect(
                host=self.selectdb_ip,
                port=9030,
                user='admin',
                password='QWer12#$',
                database=db,
                cursorclass=DictCursor,
                read_timeout=1,
                write_timeout=1,
                autocommit=False
            )

            cursor = db.cursor()

            cursor.execute(sql_str)
            operate = sql_str.split(" ")[0]
            if operate.lower() in ["select", "desc"]:
                rows = cursor.fetchall()
                if len(rows):
                    response_data['message'] = "Success"
                    response_data['data'] = rows
                    cursor.close()
                    return response_data
                else:
                    cursor.close()
                    response_data['message'] = "fail"
                    cursor.close()
                    return response_data
            elif operate.lower() in ["drop", "delete", "update", "truncate", "alter", "insert", "create"]:
                db.commit()
                response_data['message'] = "Success"
                response_data['data'] = "True"
                db.close()
                return response_data
        except Exception as msg:
            response_data['message'] = "fail"
            response_data['data'] = str(msg)
            print("starrocks 错误信息： ", msg)
        return response_data

    def exec_saphana_sql(self, db, sql_str):
        logger.info("saphana执行的sql是：{}, 对应库是 {}".format(db,sql_str))
        response_data = {'message': '', 'data': []}
        try:
            connection = dbapi.connect(address=self.sap_hana_ip,
                                        port=39017,
                                        user="SYSTEM",
                                        password="HXEHana1",
                                        db_name=db)
            cursor = connection.cursor()
            cursor.execute(sql_str)
            operate = sql_str.split(" ")[0]
            if operate.lower() in ["select", "desc", "show"]:
                rows = cursor.fetchall()
                logger.info("执行后 rows：{}".format(rows))
                if len(rows):
                    response_data['message'] = "Success"
                    response_data['data'] = rows
                    cursor.close()
                    return response_data
                else:
                    cursor.close()
                    response_data['message'] = "fail"
                    cursor.close()
                    return response_data
            elif operate.lower() in ["drop", "delete", "update", "truncate", "alter", "insert", "create"]:
                response_data['message'] = "Success"
                response_data['data'] = "True"
                cursor.close()
                return response_data
        except Exception as msg:
            response_data['message'] = "fail"
            response_data['data'] = str(msg)
            logger.info("hana 链接错误g：{}".format(msg))
        return response_data

    def exec_dolphindb_sql(self, db="whalescheduler", table="v26_dolphindb_data_drop", operate="select"):
        s = ddb.Session()
        s.connect(self.dolphindb_ip, 8848, "admin", "123456")

        dic_res = 0
        if operate in ["select"]:
            try:
                sql = "select count(*) from tmp;"
                sql = '''
                tmp = loadTable("dfs://{}", '{}');
                {}
                '''.format(db, table, sql)
                rows = s.run(sql)
                if len(rows):
                    dic_res = rows
                else:
                    dic_res = 0
            except Exception as msg:
                print("dolphindb 错误信息： ", msg)
        elif operate in ["drop", "delete", "update", "truncate", "alter", "insert"]:
            s.dropTable(dbPath="dfs://whalescheduler", tableName="{}".format(table))
        elif operate in ["createUser"]:
            username = "usera"
            password = "QWer12#$"
            sql = f"createUser('{username}', '{password}');"
            s.run(sql)
            print("用户创建成功")
        print(dic_res)
        return dic_res

    def exec_s3_sql(self):
        FORMAL_S3_CONFIG = {
            "s3_endpoint": "s3.cn-north-1.amazonaws.com.cn",
            "access_key": "AKIAYYUV5DMXNWIDLUEB",
            "secret_key": "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"
        }

        url = "https://{}".format(FORMAL_S3_CONFIG['s3_endpoint'])
        session = boto3.Session(
            aws_access_key_id=FORMAL_S3_CONFIG['access_key'],
            aws_secret_access_key=FORMAL_S3_CONFIG['secret_key']
        )
        s3_client = session.client('s3', endpoint_url=url)
        return s3_client

    def exec_hive_sql(self, db, sql_str):
        conn = hive.Connection(
            host=self.hive_ip,
            port=9803,
            database=db)
        operate = sql_str.split(" ")[0]
        cursor = conn.cursor()
        cursor.execute(sql_str)

        rows = ''
        if operate in ["select", "desc"]:
            try:
                rows = cursor.fetchall()
            except Exception as msg:
                print("hive 错误信息： ", msg)
            if len(rows):
                dic_res = rows
            else:
                dic_res = 0
            cursor.close()
        elif operate in ["drop", "delete", "update", "truncate", "alter", "insert"]:
            conn.commit()
        conn.close()
        return dic_res

    def exec_es_sql(self, db, sql=ElasticSearchOperate, version=None, res_log=None):
        """
        参考-查询：https://blog.csdn.net/feizuiku0116/article/details/122470501
        参考-删除：https://blog.csdn.net/feizuiku0116/article/details/122470474
        :param sql:
        :return:
        """
        response_data = {'message': '', 'data': []}
        try:
            operate = sql.operate
            table = sql.table
            data = sql.data

            myclient = Elasticsearch([{'host': self.es_ip, 'port': 9200, 'scheme': 'http'}])
            collection_method = getattr(myclient, operate)
            if (sql.hasindices):
                collection_method = getattr(myclient.indices, operate)

            response_data['message'] = collection_method(index=table, body=data)
            response_data['data'] = "True"
            print(response_data["message"])

            self.set_flag_db("es", True)
        except Exception as msg:
            print("es 错误信息： ", msg)
            response_data['message'] = "fail"
        return response_data

    def exec_kafka_sql(self, cmd, topic):
        print("执行的kafka 语句是：{}, 对应topic是：{}".format(cmd, topic))
        res = self.exec_cmd(cmd)
        return res

    def batch_exec_informix_sql(self, database, inser_command):
        response_data = {'message': '', 'data': []}
        insert_sql = inser_command["insert_sql"]
        values_list = inser_command["values_list"]
        jar_path = self.find_driver_jar("informix")
        print("执行批量插入")
        try:
            response_data={}
            db = jaydebeapi.connect(jclassname="com.informix.jdbc.IfxDriver",
                                    url="jdbc:informix-sqli://{}:9088/{}".format(self.informix_ip, database),
                                    driver_args=["informix", "in4mix"], jars=jar_path)
            db.jconn.setAutoCommit(False)  # 关闭自动提交以启用事务
            cursor = db.cursor()
            cursor.executemany(insert_sql, values_list)
            db.commit()
            db.jconn.setAutoCommit(True)  # 恢复自动提交
            cursor.close()
            db.close()
            response_data['message'] = "Success"
            response_data['data'] = "informix批量插入成功"
            return response_data
        except Exception as msg:
            logger.info("informix 批量插入错误：", msg)
            response_data['message'] = "fail"
            response_data['data'] = str(msg)
        return response_data


    def find_driver_jar(self, db_type):
        # 定义两个路径
        local_path = "/Users/wendada/PycharmProjects/qa_ci_cd/wt/resources/jar/"
        online_path = "/data/whale_ops/auto_test/qa_ci_cd/wt/resources/jar/"
        if db_type == "informix":
            jar_name = "informix-jdbc-complete-4.50.4.1.jar"
        elif db_type == "dameng":
            jar_name = "DmJdbcDriver18.jar"
        # 检查两个路径中哪个存在名为 informix 驱动
        if os.path.exists(os.path.join(local_path, jar_name)):
            jar_path = local_path + jar_name
        elif os.path.exists(os.path.join(online_path, jar_name)):
            jar_path = online_path + jar_name
        else:
            jar_path = None  # 或者可以设置为其他默认值，或者抛出异常
        return jar_path

    def exec_informix_sql(self, database="", sql_str="", version=None, res_log=None):
        logger.info("informix 执行sql：{}, 对应库是：{}".format(sql_str, database))
        response_data = {'message': '', 'data': []}

        jar_path = self.find_driver_jar("informix")
        logger.info("informix 加载的路径是：{}".format(jar_path))
        try:
            # 传入是字典就是批量插入
            if isinstance(sql_str, dict):
                response_data = self.batch_exec_informix_sql(database, sql_str)
                return response_data
            db = jaydebeapi.connect(jclassname="com.informix.jdbc.IfxDriver",url="jdbc:informix-sqli://{}:9088/{}".format(self.informix_ip, database),driver_args=["informix", "in4mix"],jars=jar_path)
            cursor = db.cursor()
            cursor.execute(sql_str)
            operate = sql_str.split(" ")[0]
            if version == 1:
                rows = cursor.fetchall()
                ver = "informix的版本号为：" + str(rows[0][0]) + "\n"
                res_log.write(ver.encode())
                print(ver)
            else:
                if operate.lower() in ["select", "desc", "show"]:
                    rows = cursor.fetchall()
                    logger.info("执行后 rows：{}".format(rows))
                    if len(rows):
                        response_data['message'] = "Success"
                        response_data['data'] = rows
                        cursor.close()
                        return response_data
                    else:
                        cursor.close()
                        response_data['message'] = "fail"
                        cursor.close()
                        return response_data
                elif operate.lower() in ["drop", "delete", "update", "truncate", "alter", "insert", "create"]:
                    response_data['message'] = "Success"
                    response_data['data'] = "True"
                    cursor.close()
                    return response_data
            self.set_flag_db("informix", True)
        except Exception as msg:
            logger.info("informix 链接错误g：{}".format(msg))
            response_data['message'] = "fail"
            response_data['data'] = str(msg)
        return response_data


    def exec_mongodb_sql(self, db="", operate=MongoDBOperate, version=None, res_log=None):
        response_data = {'message': '', 'data': []}
        try:
            myclient = pymongo.MongoClient("mongodb://root2:123456@" + self.mongodb_ip + ":27017")
            self.set_flag_db("mongodb", True)
            self.mongodb_client = myclient
            collection = myclient[db][operate.table]

            if operate.operate == 'create_ddl':
                collection.insert_one(operate.data)
                response_data['data'] = "True"
            else:
                collection_method = getattr(collection, operate.operate)
                response_data['message'] = collection_method(operate.data)
                response_data['data'] = "True"
            return response_data
        except Exception as msg:
            print("mongodb 链接错误： ", msg)
            response_data['message'] = "fail"
            return response_data

    def exec_starrocks_sql(self, db='', sql_str='', version=None, res_log=None):
        print("执行的starrocks语句是：{}, 对应数据库是：{}".format(sql_str, db))

        response_data = {'message': '', 'data': []}
        try:
            db = pymysql.connect(
                host=self.starrock_ip,
                port=9030,
                user='root',
                password='root',
                database=db,
                cursorclass=DictCursor,
                read_timeout=1,
                write_timeout=1,
                autocommit=False
            )

            cursor = db.cursor()

            cursor.execute(sql_str)
            operate = sql_str.split(" ")[0]
            if version == 1:
                rows = cursor.fetchall()
                ver = "starrock的版本号为：" + str(rows[0]) + "\n"
                res_log.write(ver.encode())
                print(ver)
            else:
                if operate in ["select", "desc"]:
                    rows = cursor.fetchall()
                    if len(rows):
                        response_data['message'] = "Success"
                        response_data['data'] = rows
                        cursor.close()
                        return response_data
                    else:
                        cursor.close()
                        response_data['message'] = "fail"
                        cursor.close()
                        return response_data
                elif operate in ["drop", "delete", "update", "truncate", "alter", "insert", "create"]:
                    db.commit()
                    response_data['message'] = "Success"
                    response_data['data'] = "True"
                    db.close()
                    return response_data
            self.set_flag_db("starrocks", True)
        except Exception as msg:
            dic_res = 0
            response_data['message'] = "fail"
            response_data['data'] = str(msg)
            print("starrocks 错误信息： ", msg)
        return response_data

    def exec_tidb_sql(self, db='', sql_str='', version=None, res_log=None):
        logger.info("执行的tidb语句是：{}, 对应数据库是：{}".format(sql_str, db))

        response_data = {'message': '', 'data': []}
        try:
            db = pymysql.connect(
                host=self.tidb_ip,
                port=4000,
                user='root',
                password='root@123',
                database=db,
                cursorclass=DictCursor,
                read_timeout=5,
                write_timeout=5,
                autocommit=False
            )
            cursor = db.cursor()
            cursor.execute(sql_str)
            operate = sql_str.split(" ")[0]
            rows = ''
            if version == 1:
                rows = cursor.fetchall()
                ver = "tidb的版本号为：" + str(rows[0]) + "\n"
                res_log.write(ver.encode())
                print(ver)
            else:
                if operate.lower() in ["show", "select", "desc"]:
                    rows = cursor.fetchall()
                    if len(rows):
                        response_data['message'] = "Success"
                        response_data['data'] = rows
                        cursor.close()
                        return response_data
                    else:
                        cursor.close()
                        response_data['message'] = "fail"
                        cursor.close()
                        return response_data
                elif operate.lower() in ["drop", "delete", "update", "truncate", "alter", "insert", "create"]:
                    db.commit()
                    response_data['message'] = "Success"
                    response_data['data'] = "True"
                    cursor.close()
                    return response_data
            self.set_flag_db("tidb", True)
        except Exception as msg:
            logger.info("tidb 执行错误：{}".format(msg))
            response_data['message'] = "fail"
            response_data['data'] = str(msg)
        return response_data

    def exec_s3redshift_sql(self, db='dev', sql_str=''):
        response_data = {'message': '', 'data': []}
        logger.info("执行的 s3redshift 语句是：{}, 对应db是{}".format(sql_str, db))
        try:
            dic_res = {}
            db = psycopg2.connect(
                host=self.s3redshift_ip,
                port='5439',
                database=db,
                user="awsuser",
                password="JUne..001",
                connect_timeout=5,
                options="-c search_path=public"

            )
            cursor = db.cursor()
            cursor.execute(sql_str)
            self.set_flag_db("s3redshift", True)
            rows = ''
            operate = sql_str.split(" ")[0]
            if operate.lower() in ["select", "desc"]:
                try:
                    rows = cursor.fetchall()
                    cols = [d[0] for d in cursor.description]
                    response_data['message'] = "Success"
                    response_data['data'] = rows
                except Exception as msg:
                    response_data['message'] = "fail"
                    response_data['data'] = str(msg)
                    print("s3redshift 查询错误信息： ", msg)
                for row in rows:
                    dic_res = dict(zip(cols, row))
            elif operate.lower() in ["create", "drop", "delete", "update", "truncate", "alter", "insert"]:
                db.commit()
                response_data['message'] = "Success"
                cursor.close()
                db.close()
        except Exception as msg:
            response_data['message'] = "fail"
            response_data['data'] = str(msg)
            print("s3redshift 链接错误信息： ", msg)
        return response_data

    def exec_psql_sql(self, db="engine_test", sql_str='', version=None, res_log=None):
        print("执行的psql语句是：{}, 对应db是{}".format(sql_str, db))
        response_data = {'message': '', 'data': []}

        db = psycopg2.connect(
            host=self.psql_ip,
            port='5432',
            database=db,
            user="postgres",
            password="postgres",
            connect_timeout=5,
            options="-c search_path=public"
        )

        try:
            cursor = db.cursor()
            cursor.execute(sql_str)
            operate = sql_str.split(" ")[0]
            if version == 1:
                rows = cursor.fetchall()
                ver = "pgsql的版本号为：" + str(rows[0][0]) + "\n"
                res_log.write(ver.encode())
                print(ver)
            else:
                if operate in ["select", "desc"]:
                    rows = cursor.fetchall()
                    if len(rows):
                        response_data['message'] = "Success"
                        response_data['data'] = rows
                        cursor.close()
                        return response_data
                    else:
                        cursor.close()
                        response_data['message'] = "fail"
                        cursor.close()
                        return response_data
                elif operate.lower() in ["create", "drop", "delete", "update", "truncate", "alter", "insert"]:
                    db.commit()
                    response_data['message'] = "Success"
                    response_data['data'] = "True"
                    cursor.close()
                    return response_data
            self.set_flag_db("psql", True)
        except Exception as msg:
            logger.info("psql执行错误：{}".format(msg))
            response_data['message'] = "fail"
            response_data['data'] = str(msg)
        return response_data


    def exec_highGo_sql(self, db="", sql_str='', version=None, res_log=None):
        logger.info("执行的highGO语句是：{}, 对应数据库是：{}".format(sql_str, db))
        response_data = {'message': '', 'data': []}

        db = psycopg2.connect(
            host=self.highGo_ip,
            port='5866',
            database=db,
            user="highgo",
            password="Highgo@123",
            connect_timeout=5,
            options="-c search_path=public"
        )

        try:
            cursor = db.cursor()
            cursor.execute(sql_str)
            operate = sql_str.split(" ")[0]
            if version == 1:
                rows = cursor.fetchall()
                ver = "highGO的版本号为：" + str(rows[0][0]) + "\n"
                res_log.write(ver.encode())
                print(ver)
            else:
                if operate in ["select", "desc", "show"]:
                    rows = cursor.fetchall()
                    if len(rows):
                        response_data['message'] = "Success"
                        response_data['data'] = rows
                        cursor.close()
                        return response_data
                    else:
                        cursor.close()
                        response_data['message'] = "fail"
                        cursor.close()
                        return response_data
                elif operate.lower() in ["create", "drop", "delete", "update", "truncate", "alter", "insert"]:
                    db.commit()
                    response_data['message'] = "Success"
                    response_data['data'] = "True"
                    cursor.close()
                    return response_data
            self.set_flag_db("highGo", True)
        except Exception as msg:
            logger.info("highgo执行错误：{}".format(msg))
            response_data['message'] = "fail"
            response_data['data'] = str(msg)
        return response_data

    def exec_sqlserver_sql(self, db="", sql_str='', version=None, res_log=None):
        response_data = {'message': '', 'data': []}
        logger.info("执行的sqlserver语句是：{}, 对应db是{}".format(sql_str, db))

        try:
            connection = pymssql.connect(server=self.sqls_ip + ':1433', user='sa', password='root@123', database=db)
            cursor = connection.cursor()
            cursor.execute(sql_str)
            operate = sql_str.split(" ")[0]
            if version == 1:
                rows = cursor.fetchall()
                ver = "sqlserver的版本号为：" + str(rows[0][0]) + "\n"
                res_log.write(ver.encode())
                print(ver)
            else:
                if operate in ["select", "desc", "show"]:
                    rows = cursor.fetchall()
                    if len(rows):
                        response_data['message'] = "Success"
                        response_data['data'] = rows
                        connection.close()
                        return response_data
                    else:
                        cursor.close()
                        response_data['message'] = "fail"
                        connection.close()
                        return response_data
                elif operate.lower() in ["create", "drop", "delete", "update", "truncate", "alter", "insert", "exec"]:
                    connection.commit()
                    response_data['message'] = "Success"
                    response_data['data'] = "True"
                    connection.close()
                    return response_data
            self.set_flag_db("sqls", True)
        except Exception as msg:
            logger.info("sqls执行错误：{}".format(msg))
            response_data['message'] = "fail"
            response_data['data'] = str(msg)
        return response_data

    def exec_oracle11g_sql(self, db="", sql_str='', version=None, res_log=None, execute_type='single', values=None):
        """
        执行 Oracle 11g SQL 语句。
        :param db: 数据库名称
        :param sql_str: SQL 语句或元组
        :param version: 版本信息
        :param res_log: 日志文件对象
        :param execute_type: 执行类型，'single' 或 'batch'
        :param values: 批量插入时使用的值
        :return: response_data 字典
        """
        logger.info("执行的oracle11g语句是：{}, 对应数据库是：{}".format(sql_str, db))
        if isinstance(sql_str, tuple):
            sql_str = sql_str[0]

        response_data = {'message': '', 'data': []}

        connection = cx_Oracle.connect(user="qa_source", password="qa_source", dsn=self.ora11g_ip + ":1521/orcl.test")
        cursor = connection.cursor()

        operate = sql_str.split(" ")[0].lower()

        try:
            if execute_type == 'single':
                cursor.execute(sql_str)
            elif execute_type == 'batch':
                cursor.executemany(sql_str, values)

            if version == 1:
                rows = cursor.fetchone()
                if res_log is not None:
                    res_log.write(f"oracle的版本号为：{rows[0]}\n".encode())
            else:
                if operate.lower() in ["select", "desc", "show"]:
                    rows = cursor.fetchall()
                    if len(rows):
                        response_data['message'] = "Success"
                        response_data['data'] = rows
                        connection.close()
                        return response_data
                    else:
                        cursor.close()
                        response_data['message'] = "fail"
                        connection.close()
                        return response_data
                elif operate.lower() in ["delete", "update", "truncate", "alter", "drop", "insert", "create"]:
                    connection.commit()
                    response_data['message'] = "Success"
                    response_data['data'] = "True"
                    connection.close()
                    return response_data

            self.set_flag_db("ora11g", True)

        except Exception as msg:
            response_data['data'] = str(msg)
            response_data['message'] = "fail"
            print("oracle11g 错误信息： ", msg)

        return response_data

    def exec_oracle12c_sql(self, db, sql_str='', version=None, res_log=None):
        print("执行的oracle12c语句是：{}, 对应db是{}".format(sql_str, db))
        response_data = {'message': '', 'data': []}

        # connection = cx_Oracle.connect(user="QA_SOURCE", password="oracle", dsn=self.ora12c_ip + ":1523/orcl.test")
        connection = cx_Oracle.connect(user="qa_source", password="oracle", dsn=self.ora12c_ip + ":1523/xe")

        operate = sql_str.split(" ")[0]

        try:
            cursor = connection.cursor()
            cursor.execute(sql_str)
            if version == 1:
                rows = cursor.fetchone()
                ver = "oracle的版本号为：" + str(rows[0]) + "\n"
                res_log.write(ver.encode())
                print(ver)
            else:
                if operate.lower() in ["select", "desc", "show"]:
                    rows = cursor.fetchall()
                    if len(rows):
                        response_data['message'] = "Success"
                        response_data['data'] = rows
                        connection.close()
                        return response_data
                    else:
                        cursor.close()
                        response_data['message'] = "fail"
                        connection.close()
                        return response_data
                elif operate.lower() in ["delete", "update", "truncate", "alter", "drop", "insert", "create"]:
                    connection.commit()
                    response_data['message'] = "Success"
                    response_data['data'] = "True"
                    connection.close()
                    return response_data
            self.set_flag_db("ora12c", True)
        except Exception as msg:
            print("oracle12c 错误信息： ", msg)

    def exec_sql_file(self, db='', f_path='', types=''):
        file = open(f_path, 'r', encoding='utf-8').read()
        cmds = file.rsplit(';')[:-1]
        # 巨坑的换行符, 必须要去掉否则无法全部初始化表
        cmds = [x.strip() for x in cmds]
        print(cmds)
        for c in cmds:
            try:
                if types == "mysql":
                    print("开始truncate: mysql")
                    self.exec_sql(db, c)
                elif types == "oracle":
                    print("开始truncate: ora")
                    self.exec_oracle_sql(db, c)
                elif types == "sqls":
                    print("开始truncate: sqls")
                    self.exec_sqlserver_sql(db, c)
                elif types == "psql":
                    print("开始truncate: psql")
                    self.exec_psql_sql(db, c)
                elif types == "s3redshift":
                    print("开始truncate: s3redshift")
                    self.exec_s3redshift_sql(db, c)
                elif types == "tidb":
                    print("开始truncate: tidb")
                    self.exec_tidb_sql(db, c)
                elif types == "starrocks":
                    print("开始truncate: starrocks")
                    self.exec_starrocks_sql(db, c)
                elif types == "doris":
                    print("开始truncate: doris")
                    self.exec_doris_sql(db, c)
                elif types == "kingbase":
                    print("开始truncate: kingbase")
                    self.exec_kingbase_sql(db, c)
                elif types == "kingbase":
                    print("开始truncate: kingbase")
                    self.exec_kingbase_sql(c)
                elif c == 'quit':
                    break
            except Exception as msg:
                print("错误信息： ", msg)

    def exec_doris_sql(self, db='', sql_str='', res_log=None, version=1):
        response_data = {'message': '', 'data': []}
        logger.info("执行的doris 语句是：{}, 对应db是{}".format(sql_str, db))
        try:
            if version == 1:
                port = 9031
                user = "root"
                password = "123456"
            elif version == 2:
                port = 9034
                user = "root"
                password = "root@123"
            else:
                logger.info("doris没有输入版本，请重新输入version")
            connection = pymysql.connect(host=self.doris_ip, port=port, user=user,password=password, database=db,read_timeout=5, write_timeout=5, charset='utf8')
            cursor = connection.cursor()
            cursor.execute(sql_str)
            operate = sql_str.split(" ")[0]
            if operate in ["select", "desc", "show"]:
                rows = cursor.fetchall()
                if len(rows):
                    response_data['message'] = "Success"
                    response_data['data'] = rows
                    connection.close()
                    return response_data
                else:
                    cursor.close()
                    response_data['message'] = "fail"
                    connection.close()
                    return response_data
            elif operate.lower() in ["create", "drop", "delete", "update", "truncate", "alter", "insert"]:
                connection.commit()
                response_data['message'] = "Success"
                response_data['data'] = "True"
                connection.close()
                return response_data
            self.set_flag_db("sqls", True)
        except Exception as msg:
            response_data['message'] = "fail"
            response_data['data'] = str(msg)
            print("doris 错误信息： ", msg)
        return response_data

    def set_flag_db(self, db='', flag=''):
        self.flag_db[db] = flag

    def exec_openGauss_db(self, db='', sql_str='', version=None, res_log=None):
        print("执行的openGauss语句是：{}, 对应数据库是：{}".format(sql_str, db))
        response_data = {'message': '', 'data': []}

        try:
            connection = psycopg2.connect(host=self.openGauss_ip, port=5433, user="test01", password="Bigdata@123",
                                          database="wls")

            cursor = connection.cursor()

            cursor.execute(sql_str)
            operate = sql_str.split(" ")[0]
            if version == 1:
                rows = cursor.fetchall()
                ver = "openGauss的版本号为：" + str(rows[0]) + "\n"
                res_log.write(ver.encode())
                print(ver)
            else:
                if operate in ["select", "desc", "show"]:
                    rows = cursor.fetchall()
                    if len(rows):
                        response_data['message'] = "Success"
                        response_data['data'] = rows
                        connection.close()
                        return response_data
                    else:
                        cursor.close()
                        response_data['message'] = "fail"
                        connection.close()
                        return response_data
                elif operate.lower() in ["create", "drop", "delete", "update", "truncate", "alter", "insert"]:
                    connection.commit()
                    response_data['message'] = "Success"
                    response_data['data'] = "True"
                    connection.close()
                    return response_data
            self.set_flag_db("mysql", True)
        except Exception as msg:
            print("openGauss 链接不通： ", msg)

    def exec_sql(self, db='', sql_str='', version=None, res_log=None):
        logger.info("执行的mysql语句是：{}, 对应数据库是：{}".format(sql_str, db))
        # print("执行的mysql语句是：{}, 对应数据库是：{}".format(sql_str, db))
        response_data = {'message': '', 'data': []}
        try:
            db = pymysql.connect(host=self.mysql_ip, user="root", password="root@123", db=db, read_timeout=5,
                                 write_timeout=5, cursorclass=pymysql.cursors.DictCursor, port=3306)
            cursor = db.cursor()

            cursor.execute(sql_str)
            operate = sql_str.split(" ")[0]
            if version == 1:
                rows = cursor.fetchall()
                ver = "mysql的版本号为：" + str(rows[0]) + "\n"
                res_log.write(ver.encode())
                print(ver)
            else:
                if operate.lower() in ["select", "desc", "show"]:
                    rows = cursor.fetchall()
                    if len(rows):
                        response_data['message'] = "Success"
                        response_data['data'] = rows
                        db.close()
                        print(response_data)
                        return response_data
                    else:
                        cursor.close()
                        response_data['message'] = "fail"
                        db.close()
                        return response_data
                elif operate.lower() in ["create", "drop", "delete", "update", "truncate", "alter", "insert"]:
                    db.commit()
                    response_data['message'] = "Success"
                    response_data['data'] = "True"
                    db.close()
                    return response_data
            self.set_flag_db("mysql", True)
        except Exception as msg:
            response_data['message'] = "fail"
            response_data['data'] = str(msg)
            print("mysql 链接不通： ", msg)
        return response_data

    def exec_kingbase_sql(self, cmd_str, version=None, res_log=None):
        print("执行的kingbase语句是：{}".format(cmd_str))
        try:
            if version == 1:
                res = str(self.exec_cmd(cmd_str))
                server_version = re.search(r'V\d+R\d+C\d+B\d+', res).group()
                ver = "kingbase的版本号为：" + server_version + "\n"
                res_log.write(ver.encode())
                print(ver)
            else:
                res = str(self.exec_cmd(cmd_str))
            self.set_flag_db("kingbase", True)
        except Exception as msg:
            print("kingbase 执行错误信息： ", msg)
        return res

    def exec_cmd(self, cmd_str):
        client = paramiko.SSHClient()
        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        client.connect(self.datasource01_ip, 22, username="root", password="whale123!@#", timeout=20)
        stdin, stdout, stderr = client.exec_command(cmd_str, get_pty=True)
        result = stdout.read().decode('utf-8')
        # print("{} 执行结果:{}".format(self.datasource01_ip, result))
        return result



    def exec_dameng_sql(self, db='', sql_str='', version=None, res_log=None, execute_type="single", values=None):
        jar_path = self.find_driver_jar("dameng")
        logger.info("执行的dameng语句是：{}, 对应数据库是：{}".format(sql_str, db))
        response_data = {'message': '', 'data': []}
        cursor = None
        try:
            conn = jaydebeapi.connect("dm.jdbc.driver.DmDriver", 'jdbc:dm://{}:5236/{}'.format(self.dm_ip,db),
                                      ['SYSDBA', 'SYSDBA001'], jar_path)
            cursor = conn.cursor()
            if execute_type == 'single':
                cursor.execute(sql_str)
            elif execute_type == 'batch':
                cursor.executemany(sql_str, values)
            operate = sql_str.split(" ")[0]
            if version == 1:
                rows = cursor.fetchall()
                ver = "dameng的版本号为：" + str(rows[0]) + "\n"
                res_log.write(ver.encode())
                print(ver)
            else:
                if operate.lower() in ["select", "desc", "show"]:
                    rows = cursor.fetchall()
                    print(rows)
                    if len(rows):
                        response_data['message'] = "Success"
                        response_data['data'] = rows
                        return response_data
                    else:
                        cursor.close()
                        response_data['message'] = "fail"
                        return response_data
                elif operate.lower() in ["create", "drop", "delete", "update", "truncate", "alter", "insert", "grant"]:
                    conn.commit()
                    response_data['message'] = "Success"
                    response_data['data'] = "True"
                    return response_data
                self.set_flag_db("dameng", True)
        except Exception as msg:
            response_data['message'] = "fail"
            response_data['data'] = str(msg)
            print("dameng 执行错误信息： ", msg)
        finally:
            if cursor is not None:
                cursor.close()
        return response_data

    def get_ftp_server(self, version=None, res_log=None):
        try:
            ftp_ = FTP()
            # 连接
            ftp_.connect(self.ftp_ip, 21)
            # 登录
            login_state = ftp_.login("wenjun_ruan", "QWer12#$")

            if version == 1:
                ver = "ftp的版本号为：" + ftp_.getwelcome() + "\n"
                res_log.write(ver.res_logencode())
                print("ftp登录状态为：{}".format(login_state))

            self.set_flag_db("ftpfile", True)
            return ftp_
        except Exception as msg:
            print("ftp 登录失败： ", msg)

    def get_sftp_server(self, version=None):
        try:
            transport = paramiko.Transport((self.sftp_ip, 22))
            transport.connect(username="whale_ops", password="QWer12#$")
            sftp = paramiko.SFTPClient.from_transport(transport)
            self.set_flag_db("sftpfile", True)
            if version == 1:
                print("SFTP connection successful!")
            return sftp
        except Exception as e:
            print("Error connecting to SFTP server:", e)

    def connect_all_test(self, res_log):
        self.exec_sql("qa_sink", "select VERSION()", 1, res_log)
        self.exec_oracle12c_sql("qa_sink", "SELECT * FROM v$version WHERE banner LIKE 'Oracle%'", 1, res_log)
        self.exec_oracle11g_sql("qa_sink", "SELECT * FROM v$version WHERE banner LIKE 'Oracle%'", 1, res_log)
        self.exec_psql_sql("qa_sink", "show server_version", 1, res_log)
        self.exec_highGo_sql("qa_sink", "show server_version", 1, res_log)
        self.exec_sqlserver_sql("qa_sink", "select SERVERPROPERTY('ProductVersion')", 1, res_log)
        self.exec_doris_sql("qa_sink", "select @@version_comment;", 1, res_log)
        self.exec_tidb_sql("qa_sink", "select VERSION()", 1, res_log)
        self.exec_starrocks_sql("qa_sink", "select @@version_comment;", 1, res_log)
        self.exec_mongodb_sql("qa_sink", "db.version()", 1, res_log)
        self.exec_informix_sql("wls", "select dbinfo('version', 'full')", 1, res_log)
        self.exec_kingbase_sql(
            "docker exec -it kingbase-v8r3 bash -c 'ksql -U kingbase -W qwe123 test -f ./sql/version.sql'", 1, res_log)
        self.get_ftp_server(1, res_log)
        self.get_sftp_server(1)

        flag_db = []
        tmp = self.flag_db
        for k in tmp:
            if tmp[k]:
                flag_db.append(k)
        res = str("====== 当前畅通的db 有：{}".format(flag_db) + "\n\r").encode()
        res_log.write(res)
        return flag_db


if __name__ == '__main__':
    ba = DataConnect()
    # ba.exec_saphana_sql("qa_source","CREATE TABLE T2 (ID INTEGER, NAME NVARCHAR(10))")
    # ba.exec_saphana_sql("qa_source", "INSERT INTO T1 (ID, NAME) VALUES (1, 'Name1')")
    ba.exec_selectdb_sql("qa_source", "select 1")

    # ba.exec_dameng_sql("QA_SOURCE", "drop table if exists QA_SOURCE.CDC_DAMENG_DML_T1")
